#pragma once

#include <functional>
#include <memory>
#include <string>
#include <unordered_map>
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>

#include <boost/date_time/posix_time/posix_time.hpp>

#include "mysql_outbox.hpp"
#include "rabbitmq.hpp"
#include "data_redis.hpp"
#include "data_es.hpp"

namespace suye
{
struct DispatchContext
{
    std::shared_ptr<sw::redis::Redis> redis;
    std::shared_ptr<elasticlient::Client> es;
    MQClient::ptr mq;
};

class OutboxDispatchWorker
{
public:
    using ptr = std::shared_ptr<OutboxDispatchWorker>;
    using Handler = std::function<bool(const std::shared_ptr<OutboxEvent>&, const DispatchContext&)>;

    struct Options
    {
        std::size_t batch_size{32};
        boost::posix_time::time_duration dispatch_timeout{boost::posix_time::seconds(30)};
        boost::posix_time::time_duration requeue_timeout{boost::posix_time::minutes(1)};
        boost::posix_time::time_duration poll_interval{boost::posix_time::seconds(2)};
    };

    OutboxDispatchWorker(const OutboxTable::ptr& table,
                         const DispatchContext& context,
                         const Options& options = Options());

    ~OutboxDispatchWorker();

    void register_handler(const std::string& event_type, const Handler& handler);

    void start();
    void stop();

    void process_once();

private:
    void run();

private:
    OutboxTable::ptr _table;
    DispatchContext _context;
    Options _options;

    std::unordered_map<std::string, Handler> _handlers;

    std::atomic<bool> _running{false};
    std::thread _worker;
    std::mutex _mutex;
    std::condition_variable _cv;
};
}
